package mr

import "log"
import "net"
import "os"
import "net/rpc"
import "net/http"
// import "errors"
import "time"
import "sync"
import "fmt"
// type Worker mr.Worker
type WorkerType int
type TaskStatus int
type TaskType   int


const FREE WorkerType = 0
const BUSY WorkerType = 1
const DIE  WorkerType = 2

const READY   TaskStatus = 0
const RUNNING TaskStatus = 3
const FINISH  TaskStatus = 1
const FAIL    TaskStatus = 2

const MAP             TaskType = 0
const REDUCE          TaskType = 1
const SHUTDOWN        TaskType = 2
const EMPTY           TaskType = 3

var workerIdChan chan int

type WorkerInfo struct {
    id          int
    member      int
    last_beat   int64
    status      WorkerType
    tasks       []*TaskInfo
}

func (this *WorkerInfo) addTask(info *TaskInfo) {
    this.tasks = append(this.tasks, info)
}

// func (this *WorkerInfo) removeTask(taskId int) {
//     fmt.Println("remove task:", taskId)
//     for i, t := range this.tasks {
//         if t.Id == taskId {
//             this.tasks[i] = this.tasks[len(this.tasks)-1]
//             this.tasks = this.tasks[:len(this.tasks)-1]
//             break
//         }
//     }
// }

type TaskInfo struct {
    Id int
    Member int
    Kind TaskType
    Resources []string
    Status TaskStatus
    
    // lock      sync.Mutex
}


type Coordinator struct {
    // Your definitions here.
    nReduce       int
    tasks         []*TaskInfo
    workers       map[int]*WorkerInfo
    MemberCount   int
    mapTasks      []*TaskInfo
	reduceTasks   []*TaskInfo
    mu      sync.Mutex

    MatrixResources   [][]string

    doneCh        chan bool

    level         int  // 0 执行map   1 执行reduce
    totalTask     int
    todoTask      int
    doingTask     int
}



// 指派任务信息到注册的服务信息中
func (c *Coordinator) assignment(workerId int, taskInfo *TaskInfo) {
    c.workers[workerId].addTask(taskInfo)
    c.doingTask++
    c.todoTask--
    taskInfo.Status = RUNNING
}

// map/reduce任务
func (c *Coordinator) newTask(info *TaskInfo)  *Task {
    return &Task {
        Kind : info.Kind,
        Resources     : info.Resources,
        Member : info.Member,
    }
}

// worker停止任务
func (c *Coordinator) newShutdownTask() *Task {
    return &Task {
        Kind: SHUTDOWN,
    }
}

// 空任务
func (c *Coordinator) emptyTask() *Task {
    return &Task {
        Kind: EMPTY,
    }
}

func (c *Coordinator) taskFinished() bool {
    return c.doingTask == 0 && c.todoTask == 0
}

func (c *Coordinator) mapLevel() bool{
    return c.level == 0
}

func (c *Coordinator) reduceLevel() bool{
    return !c.mapLevel()
}

func (c *Coordinator) initStatistics() {
    c.totalTask = len(c.tasks)
    c.todoTask = c.totalTask
    c.doingTask = 0
}

func (c *Coordinator) syncGetTask(workerId int) *Task {
    c.mu.Lock()
    defer c.mu.Unlock()

    // if worker已失效
    if !c.isRegistered(workerId) {
        return c.newShutdownTask()
    }

    if c.taskFinished() {
        if c.mapLevel() {
            c.tasks = c.martix2ReducsTask()
            c.initStatistics()
            c.level = 1
        } else {
            return c.newShutdownTask()
        }
    }

    for _, t := range c.tasks {
        if(t.Status == READY) {
            c.assignment(workerId, t)
            return c.newTask(t)
        }
    }
    
    return c.emptyTask()

}

func (c *Coordinator) isRegistered(id int) bool {
    if _, ok := c.workers[id]; !ok {
        return false
    }
    return true
}

func (c *Coordinator) GetTask(args *GetTaskReq, reply *GetTaskResp) error {
    // todo worker的有效性验证
    task := c.syncGetTask(args.WorkerId)
    reply.Data = *task
    return nil
}


func (c *Coordinator) CompleteTask(args *CompleteTaskReq, reply *CompleteTaskResp) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    taskId := args.Id
    if !c.isRegistered(args.WorkerId) {
        reply.Status = 1
        return nil
    }

    // 修改task状态
    worker := c.workers[args.WorkerId]
    for _, t := range worker.tasks {
        if t.Id == taskId {
            t.Status = FINISH
        }
    }
    worker.tasks = make([]*TaskInfo, 0)
    // 矩阵 + path
    if args.Kind == MAP {
        c.MatrixResources[args.Member] = args.Resources
    }

   // 修改面板状态
    c.doingTask--
    if c.reduceLevel() && c.taskFinished() {
        c.doneCh <- true
    }
    reply.Status = 0
    // c.printPanel()
    return nil
}

func (c *Coordinator) Register(args *RegisterArgs, reply *RegisterReply) error {
    // workerId := args.Id
    // if _, ok := c.workers[workerId]; ok {
    //     return errors.New("exists a same worker id or already regist. id: " + strconv.Itoa(workerId))
    // }
    // todo
    c.mu.Lock()
    defer c.mu.Unlock()
    workerId := <- workerIdChan
    worker := &WorkerInfo {
        id:     workerId,
        member:     c.MemberCount,
        last_beat:  time.Now().Unix(),
        status:     FREE,
    }
    c.workers[workerId] = worker
    log.Printf("a worker [%d] regist success.", workerId)
    c.MemberCount = c.MemberCount + 1
    data := RegisterData {
        Member:         worker.member,
        Heart_interval: 1,
		Nreduce:        c.nReduce,
    }
    reply.Data = data
    reply.Id = workerId
    
    return nil
}


func (c *Coordinator) Alive(args *KeepAliveArgs, reply *KeepAliveReply) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    id := args.Id
    if !c.isRegistered(id) {
        reply.Success = 1
        return nil
    }

    worker := c.workers[id]
    // log.Printf("worker %d is alive.", id)
    worker.last_beat = time.Now().Unix()
    return nil
}


func (c *Coordinator) printPanel() {
    fmt.Println("taskLevel:", c.level)
    fmt.Println("totalTask:", c.totalTask)
    fmt.Println("todoTask:", c.todoTask)
    fmt.Println("doingTask:", c.doingTask)
}


func (c *Coordinator) martix2ReducsTask() []*TaskInfo {
    taskInfos := make([]*TaskInfo, 0)
    
    for i := 0; i < c.nReduce; i++ {
        taskInfos = append(taskInfos, &TaskInfo {
            Member: i,
            Status: READY,
            Kind: REDUCE,
        })
    }

    for _, resources := range c.MatrixResources {
        for j, resource := range resources {
            taskInfos[j].Resources = append(taskInfos[j].Resources, resource)
        }
    }
    return taskInfos
}


// start a thread that listens for RPCs from worker.go
//
func (c *Coordinator) server() {
    rpc.Register(c)
    rpc.HandleHTTP()

    sockname := coordinatorSock()
    os.Remove(sockname)
    l, e := net.Listen("unix", sockname)
    if e != nil {
        log.Fatal("listen error:", e)
    }
    go http.Serve(l, nil)
}

// 定时调用Done函数
// main/mrcoordinator.go calls Done() periodically to find out
// if the entire job has finished.
//
func (c *Coordinator) Done() bool {
    isDone := <- c.doneCh
    fmt.Println("master is close.")
    return isDone
}

// 分割源文件为多个task，这里每个文件为一个task
func createMapTasks(files []string) []*TaskInfo {
    tasks := []*TaskInfo{}
    for i,filePath := range files {
        t := &TaskInfo {
            Kind:       MAP,
            Status:     READY,
            Resources:      []string{filePath},
            Member:     i,
        }
        tasks = append(tasks, t)
    }
    return tasks;
}


func (c *Coordinator) workerActiveSchedule(expire int, interval int) {
    for {
        time.Sleep(time.Duration(interval) * time.Second)
        c.mu.Lock()
        cur := time.Now().Unix()
        deads := make([]int, 0)
        for id, worker := range c.workers {
            lastTime := worker.last_beat
            if cur - lastTime >= int64(expire) {
                for _, task := range worker.tasks {
                    task.Status = READY
                }
                deads = append(deads, id)
            }
        }

        for _, id := range deads {
            delete(c.workers, id)
        }
        c.todoTask += len(deads)
        c.doingTask -= len(deads)
        c.mu.Unlock()

    }

}

//
// create a Coordinator.
// main/mrcoordinator.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeCoordinator(files []string, nReduce int) *Coordinator {
    
    resources := make([][]string, len(files))
    for i := 0; i < len(resources); i++ {
        resources[i] = make([]string, nReduce)
    }

    c := Coordinator{
        mapTasks:      []*TaskInfo{},
        reduceTasks:   []*TaskInfo{},
        nReduce:       nReduce,
        workers:       make(map[int]*WorkerInfo),
        MemberCount:   0,  
        MatrixResources:   resources,
        doneCh:        make(chan bool),
    }
    c.level = 0
    c.doingTask = 0
    c.mapTasks = createMapTasks(files)

    c.tasks = c.mapTasks
    c.totalTask = len(c.tasks)
    c.todoTask = c.totalTask

    workerIdChan = make(chan int)
    go func() {
        workerCount := 0
        for {
            workerCount += 1 
            workerIdChan <- workerCount
        }
    }()

    go c.workerActiveSchedule(10, 10)

    c.server()
    
    return &c
}
